local newt = reload("delt")
local delt = reload("delt")
local class = reload("class")
local super = reload("hiredisLC")
local cmsgpack = require("cmsgpack")
local logerror = reload("single.logerror")

---@class xstream redis-stream使用
local this = class(super)

---构造函数(私有)
function this:ctor(name, info)
    super.ctor(self, name, info)
end

---协程处理(私有)
function this:proc(carg)
    ---建立连接
    super.launch(self)({
        auth = self:comark(carg)
    })
    self:await()

    ---创建队列
    self:xnewMsg(carg.key, carg.groupname)(self.coback, self, carg)
    self:await()

    ---轮询读取
    while (not carg.exit) and (not cgzShutdown) do
        ---读取消息
        self:xgetMsg(carg.num)(self.coback, self, carg)
        local arr, ids = self:await()
        ---确认消息
        self:xackMsg(ids)

        ---消费消息
        for _, msg in ipairs(arr) do
            self:consume(carg, msg)
            delt(msg)
        end

        ---回收数据
        delt(arr)
        delt(ids)
    end
end

---消息消费(私有)
function this:consume(carg, msg)
    local list = carg.reg[msg.cmd]
    if not list then
        logerror("未知消息:", msg)
        return
    end

    local info = cmsgpack.unpack(msg.pck)
    for _, that in ipairs(list) do
        that:xmsg(msg.fid, msg.cmd, msg.pck)
    end
    delt(info)
end

---启动函数(公有)
function this:launch(key, num)
    local carg = {
        key = key, ---消息队列
        num = num, ---读取数量
        reg = {},  ---注册消息
    }
    self:async(self.proc, carg)
end

---注册监听(公有)
function this:regitCall(cmd, that)
    local reg = self.carg.reg
    local list = reg[cmd] or newt()
    reg[cmd] = list

    table.insert(list, that)
end

---发送消息(公有)
---@param key string @消息队列名称
---@param cmd number @消息号标识
---@param info table @消息内容
function this:xaddMsg(key, cmd, info)
    local map = {
        fid = cgzSid,
        cmd = cmd,
        pck = cmsgpack.pack(info)
    }
    return super.xaddMsg(self, map, key)
end

return this

--[[
    msg = {
        fid = 发送服务器标识
        cmd = 发送消息号标识
        pck = 发送消息二进制
    }
]]
